const zmq = require('zeromq');
const fs = require('fs');
const path = require('path');
const { MessageBroker } = require('./index.js');

class ZeroMQBroker extends MessageBroker {
    constructor() {
        super();
        this.router = new zmq.Router(); // 用于接收 worker 的请求和返回信息
        this.queues = new Map(); // 存储消息队列
        this.filePath = path.join(__dirname, 'messages.json');
        this.loadMessages();
        this.handlers = new Map();
        this.workers = new Map(); // 存储消费者状态
        this.heartbeatInterval = 5000; // 心跳间隔时间
    }

    // 加载持久化的消息
    loadMessages() {
        if (fs.existsSync(this.filePath)) {
            const data = fs.readFileSync(this.filePath, 'utf8');
            this.queues = new Map(JSON.parse(data));
        }
    }

    // 保存消息到文件
    saveMessages() {
        fs.writeFileSync(this.filePath, JSON.stringify([...this.queues]), 'utf8');
    }

    // 启动 Broker
    async start(zmqUrl = 'tcp://0.0.0.0:5555') {
        await this.router.bind(zmqUrl);
        console.log(`## ZeroMQ Broker 已启动: ${zmqUrl}...`);

        // 将消息处理逻辑放到异步任务中
        this.processMessages();
        this.startHeartbeatCheck();

        // 返回一个 Promise，表示 Broker 已启动
        return Promise.resolve();
    }

    // 异步处理消息
    async processMessages() {
        for await (const message of this.router) {
            try {
                // 解析多帧消息
                const [workerId, content] = message;
                const workerIdHex = workerId.toString('hex');
                // console.log(`workerId = `, workerIdHex, ', content = ', content.toString());

                if (!workerId) {
                    console.error('无效的 workerId');
                    continue;
                }
                const { taskType, action, payload } = JSON.parse(content.toString());
                if (action === 'requestTask') {
                    // 处理 worker 的请求任务
                    if (this.queues.has(taskType) && this.queues.get(taskType).length > 0) {
                        const task = this.queues.get(taskType).shift();
                        this.router.send([workerId, JSON.stringify({ taskType, task })]);
                    } else {
                        this.router.send([workerId, JSON.stringify({ taskType, task: null })]);
                    }
                } else if (action === 'submitResult') {
                    // 处理 worker 的返回结果
                    const handlers = this.handlers.get(taskType) || [];
                    handlers.forEach((handler) => handler(payload));
                } else if (action === 'heartbeat') {
                    // 更新消费者状态
                    this.workers.set(workerIdHex, {
                        address: payload.ip,
                        lastHeartbeat: Date.now(),
                        customInfo: payload.customInfo
                    });
                }
            } catch (error) {
                console.error('处理消息时出错:', error);
            }
        }
    }

    // 外部pipeline引擎调用，发布消息
    publish(taskType, worker, message) {
        // console.log(`zmq publish taskType = ${taskType}, worker = ${worker}, message = `, message);
        const channelName = `${taskType}:${worker}`
        if (!this.queues.has(channelName)) {
            this.queues.set(channelName, []);
        }
        const queue = this.queues.get(channelName);
        queue.push(message);
        this.saveMessages();
        console.log(`消息已发布到队列 ${channelName}, message = \n`, this.truncateMessage(message));
    }

    // 订阅任务类型（模拟行为）
    subscribe(taskType, callback) {
        try {
            const handlers = this.handlers.get(taskType) || [];
            if (!handlers.includes(callback)) {
                handlers.push(callback);
                this.handlers.set(taskType, handlers);
            }
        } catch (error) {
            console.error('订阅频道时出错:', error);
        }
    }

    unsubscribe(taskType, callback) {
        try {
            if (this.handlers.has(taskType)) {
                const handlers = this.handlers.get(taskType);
                const index = handlers.indexOf(callback);
                if (index !== -1) {
                    handlers.splice(index, 1);
                    this.handlers.set(taskType, handlers);
                    console.log(`已取消订阅任务类型: ${taskType}`);
                } else {
                    console.warn(`未找到与任务类型 ${taskType} 关联的回调函数`);
                }
            } else {
                console.warn(`任务类型 ${taskType} 未被订阅`);
            }
        } catch (error) {
            console.error('取消订阅时出错:', error);
        }
    }

    // 启动心跳检查
    startHeartbeatCheck() {
        setInterval(() => {
            const now = Date.now();
            for (const [workerId, workerInfo] of this.workers) {
                if (now - workerInfo.lastHeartbeat > this.heartbeatInterval * 2) {
                    this.workers.delete(workerId);
                    console.log(`消费者 ${workerId} 已断开连接`);
                }
            }
        }, this.heartbeatInterval);
    }

    // 关闭 Broker
    async close() {
        this.router.close();
        this.queues.clear();
        this.saveMessages();
        console.log('## Close ZeroMQ Broker OK!');
    }
}

module.exports = { ZeroMQBroker };
